package com.study.study.rabbitmq;

import cn.hutool.core.thread.NamedThreadFactory;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.MessageProperties;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.*;

/**
 * 生产者发布消息
 *
 * @author: zack
 * @date: 2020/2/11 0011 下午 7:22
 * @Description:
 */
public class Sender {

    private static final Logger LOGGER = LoggerFactory.getLogger(Sender.class);

    private static final ThreadLocal<Channel> channelThreadLocal = new ThreadLocal<Channel>();

    private static final ThreadPoolExecutor poolExecutor = new ThreadPoolExecutor(4,
            8 - 1,
            60L,
            TimeUnit.SECONDS,
            new ArrayBlockingQueue<>(1000),
            new NamedThreadFactory("load-thread-", false),
            new ThreadPoolExecutor.CallerRunsPolicy()
    );

    public static void main(String[] args) throws IOException, TimeoutException {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("192.168.0.15");
        factory.setPort(5672);
        factory.setUsername("root");
        factory.setPassword("Zy200010!");
        factory.setConnectionTimeout(6000);
        factory.setSharedExecutor(Executors.newFixedThreadPool(4, new NamedThreadFactory("mq", false)));

        for (int i = 0; i < 5; i++) {

            poolExecutor.execute(() -> {
                try {
                    for (int j = 0; j < 10000000; j++) {
                        Channel channel = null;
                        if ((channel = channelThreadLocal.get()) == null) {
                            Connection connection = factory.newConnection();
                            LOGGER.info("当前线程{}创建", Thread.currentThread().getName());
                            channel = connection.createChannel();
                            channelThreadLocal.set(channel);
                        }

                        String str = Thread.currentThread().getName() + "lihuan" + j;
//                        channel.txSelect();
                        channel.basicPublish("storage_event", "event.storage.1.*", true, MessageProperties.PERSISTENT_TEXT_PLAIN, str.getBytes(StandardCharsets.UTF_8));
//                        channel.txCommit();
                    }
                } catch (Exception e) {
                    LOGGER.error(e.getMessage(), e);
                }
            });
        }
    }
}
